package com.raylu.bootstrap

import com.alibaba.fastjson.{JSONArray, JSONObject}
import com.raylu.util.{MyKafkaUtil, MySqlUtil, PropertiesUtil, RedisUtil}
import redis.clients.jedis.Jedis

import java.{lang, util}
import java.util.Properties

/**
 *
 * Description:
 *
 * Create by lucienoz on 2021/11/28.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
object BootStrapDimApp {
	def main(args : Array[String]) : Unit = {
		val properties : Properties = PropertiesUtil.load ( "config.properties" )
		//Get Dim Table List
		val jedis : Jedis = RedisUtil.getJedisFromPool ()
		jedis.select(properties.getProperty("redis.dimtable.db").toInt)
		val dimTable : util.Set[String] = jedis.smembers("dimtable")

		dimTable.forEach(tableName=>{
			val sql : String = s"select * from gmall.${tableName}"
			val jSONObjects : List[JSONObject] = MySqlUtil.queryList ( sql )
			for (elem : JSONObject <- jSONObjects) {
				val canalJSONObj = new JSONObject ()
				val dataJSONArr = new JSONArray ()
				dataJSONArr.add(elem)
				canalJSONObj.put("data",dataJSONArr)
				canalJSONObj.put("database","gmall")
				canalJSONObj.put("id",elem.getIntValue("id"))
				canalJSONObj.put("table",tableName)
				canalJSONObj.put("type", "INSERT")

				MyKafkaUtil.send(properties.getProperty("kafka.db.topic"),elem.getString("id"),canalJSONObj.toJSONString)
			}

			MyKafkaUtil.flush()



		})


	}

}
